package com.aliothservice.catfish.device;

import java.text.DateFormat;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.time.LocalDateTime;
import java.util.Date;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.geo.GeoJsonPoint;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.data.redis.core.RedisTemplate;

import com.aliothservice.catfish.admin.data.PlatformRealtimeInfo;
import com.aliothservice.catfish.device.data.DeviceOrder;
import com.aliothservice.catfish.device.data.HumidityDocument;
import com.aliothservice.catfish.device.data.HumidityObject;
import com.aliothservice.catfish.device.data.LocationDocument;
import com.aliothservice.catfish.device.data.LocationObject;
import com.aliothservice.catfish.device.data.TemperatureDocument;
import com.aliothservice.catfish.device.data.TemperatureObject;
import com.mongodb.WriteResult;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

public class CollectorInboundAdapter extends ChannelInboundHandlerAdapter {
	
	public static final int HEADER_LENGTH = 16;
	
	public static final short CMD_HEARTBEAT = 0;
	public static final short CMD_REPORTING_LOCATION = 1;
	public static final short CMD_REPORTING_TEMPERATURE = 2;
	public static final short CMD_REPORTING_HUMIDITY = 3;
	
	public static final short RES_CMD_KEEP_GOING = 0;
	public static final short RES_CMD_START_COLLECTING = 1;
	public static final short RES_CMD_STOP_COLLECTING = 2;
	
	public static final short RES_STATUS_ACCEPTED = 0;
	public static final short RES_STATUS_NOT_PROCESSED = 1;
	
	private ByteBuf headerBuffer;
	private ByteBuf bodyBuffer;
	
	private boolean hasHeaderRead = false;
	private String deviceSN = null;
	private int bodyLength = -1;
	private short command = -1;
	
	@Override
	public void handlerAdded(ChannelHandlerContext context) {
		// allocate header buffer
		headerBuffer = context.alloc().buffer(HEADER_LENGTH);
	}
	
	@Override
	public void handlerRemoved(ChannelHandlerContext context) {
		headerBuffer.release();
		headerBuffer = null;
		if (null != bodyBuffer) {
			bodyBuffer.release();
			bodyBuffer = null;
		}
	}

	@Override
	public void channelRead(ChannelHandlerContext context, Object message) {
		// mock up one order
//		if (null == redisTemplate.opsForValue().get("A0001888")) {
//			DeviceOrder mockUp = new DeviceOrder();
//			mockUp.setDeviceSN("A0001888");
//			mockUp.setType(DeviceOrder.TYPE_VEHICLE);
//			mockUp.setTemperatureMin(Integer.valueOf(2));
//			mockUp.setTemperatureMax(Integer.valueOf(8));
//			mockUp.setOrderId("fake_order_id");
//			mockUp.setLicensePlate("京N2CX01");
//			redisTemplate.opsForValue().set("A0001888", mockUp);
//		}
//		if (null == redisTemplate.opsForValue().get("A0002888")) {
//			DeviceOrder mockUp = new DeviceOrder();
//			mockUp.setDeviceSN("A0002888");
//			mockUp.setType(DeviceOrder.TYPE_VEHICLE);
//			mockUp.setTemperatureMin(Integer.valueOf(2));
//			mockUp.setTemperatureMax(Integer.valueOf(8));
//			mockUp.setOrderId("fake_order_id");
//			mockUp.setLicensePlate("京N2CX02");
//			redisTemplate.opsForValue().set("A0002888", mockUp);
//		}
		
		ByteBuf buffer = (ByteBuf)message;
		DeviceOrder order = null;
		if (!hasHeaderRead) {
			if (logger.isDebugEnabled()) {
				logger.debug("reading from client: " + context.channel().remoteAddress().toString());
			}
			// check if header is received
			if (buffer.readableBytes() < HEADER_LENGTH) {
				return;
			}
			headerBuffer.writeBytes(buffer, HEADER_LENGTH);

			// dumping out request
			byte[] snBytes = new byte[8];
			headerBuffer.readBytes(snBytes);
			deviceSN = new String(snBytes);
			bodyLength = headerBuffer.readInt();
			command = headerBuffer.readShort();
			logger.info("got request - deviceSN: " + deviceSN + " command: " + command + " body length: " + bodyLength);
			hasHeaderRead = true;

			// safty check
			if (bodyLength > 1024*1024 /* 1M */) {
				// invalid request, echo header
				logger.info("body too big, rejected, deviceSN: " + deviceSN);
				buffer.skipBytes(buffer.readableBytes());
				final ByteBuf response = context.alloc().buffer(8);
				response.writeShort(RES_CMD_STOP_COLLECTING);
				response.writeShort(RES_STATUS_NOT_PROCESSED);
				response.writeInt(0); // no body
				final ChannelFuture writeFuture = context.writeAndFlush(response);
				writeFuture.addListener(ChannelFutureListener.CLOSE);
				buffer.release();
				buffer = null;
				return;
			}

			// check device order
			order = (DeviceOrder)(redisTemplate.opsForValue().get(deviceSN));
			if (null == order) {
				// check command
				if (command == CollectorInboundAdapter.CMD_HEARTBEAT) {
					// keep heart-beat
					logger.info("keep heart-beat for " + deviceSN);
					buffer.skipBytes(buffer.readableBytes());
					final ByteBuf response = context.alloc().buffer(8+8); // header + datetime
					response.writeShort(RES_CMD_KEEP_GOING);
					response.writeShort(RES_STATUS_ACCEPTED);
					response.writeInt(8); // datetime body
					LocalDateTime currentDateTime = LocalDateTime.now();
					response.writeByte(0); // discard first byte
					response.writeShort(currentDateTime.getYear());
					response.writeByte(currentDateTime.getMonthValue());
					response.writeByte(currentDateTime.getDayOfMonth());
					response.writeByte(currentDateTime.getHour());
					response.writeByte(currentDateTime.getMinute());
					response.writeByte(currentDateTime.getSecond());
					final ChannelFuture writeFuture = context.writeAndFlush(response);
					writeFuture.addListener(ChannelFutureListener.CLOSE);
				} else {
					// it is possible that any request comes in and we stop reporting immediately, recordingEnd may get updated, but request data will not take effect
					// if the device handles the stopping properly, recordingEnd will not be updated
					logger.info("stop reporting for " + deviceSN);
					buffer.skipBytes(buffer.readableBytes());
					final ByteBuf response = context.alloc().buffer(8);
					response.writeShort(RES_CMD_STOP_COLLECTING);
					response.writeShort(RES_STATUS_ACCEPTED);
					response.writeInt(0); // no body
					final ChannelFuture writeFuture = context.writeAndFlush(response);
					writeFuture.addListener(ChannelFutureListener.CLOSE);
				}
				buffer.release();
			} else {
				// process collection request
				if (command == CollectorInboundAdapter.CMD_HEARTBEAT) {
					// start reporting
					logger.info("start reporting for " + deviceSN + " temperature range: " + order.getTemperatureMin() + "~" + order.getTemperatureMax());
					buffer.skipBytes(buffer.readableBytes());
					final ByteBuf response = context.alloc().buffer(8+8+8); // header + data + datetime
					response.writeShort(RES_CMD_START_COLLECTING);
					response.writeShort(RES_STATUS_ACCEPTED);
					response.writeInt(8);
					response.writeInt(order.getTemperatureMin().intValue());
					response.writeInt(order.getTemperatureMax().intValue());
					LocalDateTime currentDateTime = LocalDateTime.now();
					response.writeByte(0); // discard first byte
					response.writeShort(currentDateTime.getYear());
					response.writeByte(currentDateTime.getMonthValue());
					response.writeByte(currentDateTime.getDayOfMonth());
					response.writeByte(currentDateTime.getHour());
					response.writeByte(currentDateTime.getMinute());
					response.writeByte(currentDateTime.getSecond());
					
					final ChannelFuture writeFuture = context.writeAndFlush(response);
					writeFuture.addListener(ChannelFutureListener.CLOSE);
					buffer.release();

					// create recording documents as necessary
					Date now = new Date();
					Query humidityQuery = new Query();
					Criteria humidityIdCriteria = Criteria.where("orderId").is(order.getOrderId());
					humidityQuery.addCriteria(humidityIdCriteria);
					if (!mongoTemplate.exists(humidityQuery, HumidityDocument.class)) {
						HumidityDocument humidityDoc = new HumidityDocument();
						humidityDoc.setOrderId(order.getOrderId());
						humidityDoc.setViewers(order.getViewers());
						humidityDoc.setRecordingStart(now);
						mongoTemplate.save(humidityDoc);
					}
					Query temperatureQuery = new Query();
					Criteria temperatureIdCriteria = Criteria.where("orderId").is(order.getOrderId());
					temperatureQuery.addCriteria(temperatureIdCriteria);
					if (!mongoTemplate.exists(temperatureQuery, TemperatureDocument.class)) {
						TemperatureDocument temperatureDoc = new TemperatureDocument();
						temperatureDoc.setOrderId(order.getOrderId());
						temperatureDoc.setViewers(order.getViewers());
						temperatureDoc.setRecordingStart(now);
						temperatureDoc.setTemperatureMin(order.getTemperatureMin());
						temperatureDoc.setTemperatureMax(order.getTemperatureMax());
						mongoTemplate.save(temperatureDoc);
					}
					Query locationQuery = new Query();
					Criteria locationIdCriteria = Criteria.where("orderId").is(order.getOrderId());
					locationQuery.addCriteria(locationIdCriteria);
					if (!mongoTemplate.exists(locationQuery, LocationDocument.class)) {
						LocationDocument locationDoc = new LocationDocument();
						locationDoc.setOrderId(order.getOrderId());
						locationDoc.setViewers(order.getViewers());
						locationDoc.setRecordingStart(now);
						mongoTemplate.save(locationDoc);
					}
				} else if (command == CollectorInboundAdapter.CMD_REPORTING_LOCATION) {
					// recording locations
					if (buffer.readableBytes() < bodyLength) {
						return;
					}
					proceedWithLocationRequestBody(context, buffer, order);
				} else if (command == CollectorInboundAdapter.CMD_REPORTING_TEMPERATURE) {
					// recording temperature
					if (buffer.readableBytes() < 28 /* integer(temperature) double(longitude) double(latitude) long(timestamp) */) {
						return;
					}
					proceedWithTemperatureRequestBody(context, buffer, order);
				} else if (command == CollectorInboundAdapter.CMD_REPORTING_HUMIDITY) {
					// recording humidity
					if (buffer.readableBytes() < 28 /* integer(humidity) double(longitude) double(latitude) long(timestamp)*/) {
						return;
					}
					proceedWithHumidityRequestBody(context, buffer, order);
				} else {
					// invalid request, echo header
					final ChannelFuture writeFuture = context.writeAndFlush(buffer);
					writeFuture.addListener(ChannelFutureListener.CLOSE);
				}
			}
		} else {
			if (command == CollectorInboundAdapter.CMD_REPORTING_LOCATION) {
				// recording locations
				if (buffer.readableBytes() < bodyLength) {
					return;
				}
				proceedWithLocationRequestBody(context, buffer, order);
			} else if (command == CollectorInboundAdapter.CMD_REPORTING_TEMPERATURE) {
				// recording temperature
				if (buffer.readableBytes() < 28 /* integer(temperature) double(longitude) double(latitude) long(timestamp) */) {
					return;
				}
				proceedWithTemperatureRequestBody(context, buffer, order);
			} else if (command == CollectorInboundAdapter.CMD_REPORTING_HUMIDITY) {
				// recording humidity
				if (buffer.readableBytes() < 28 /* integer(humidity) double(longitude) double(latitude) long(timestamp)*/) {
					return;
				}
				proceedWithHumidityRequestBody(context, buffer, order);
			}
		}
	}
	
	@Override
	public void exceptionCaught(ChannelHandlerContext context, Throwable cause) {
		if (logger.isDebugEnabled()) {
			logger.debug("exception caught - " + cause.getMessage().toString());
		}
		context.close();
	}
	
	public CollectorInboundAdapter(RedisTemplate<String, Object> redisTemplate, MongoTemplate mongoTemplate) {
		this.redisTemplate = redisTemplate;
		this.mongoTemplate = mongoTemplate;
	}
	
	protected void proceedWithLocationRequestBody(ChannelHandlerContext context, ByteBuf buffer, DeviceOrder order) {
		bodyBuffer = context.alloc().buffer(bodyLength);
		bodyBuffer.writeBytes(buffer, bodyLength);
		if (0 != bodyLength%24 /* double(longitude) double(latitude) long(timestamp) */) {
			logger.info("invalid package, bodyLength%24=" + bodyLength%24 + " connection dropped");
			buffer.skipBytes(buffer.readableBytes());
			final ByteBuf response = context.alloc().buffer(8);
			response.writeShort(RES_CMD_STOP_COLLECTING);
			response.writeShort(RES_STATUS_NOT_PROCESSED);
			response.writeInt(0); // no body
			final ChannelFuture writeFuture = context.writeAndFlush(response);
			writeFuture.addListener(ChannelFutureListener.CLOSE);
			buffer.release();
			return;
		}
		
		Query query = locationDocumentByDeviceOrder(order);
		LocationDocument locationDocument = mongoTemplate.findOne(query, LocationDocument.class);
		if (null == locationDocument) {
			logger.warn("invalid package, location document not created, connection dropped!!!");
			buffer.skipBytes(buffer.readableBytes());
			final ByteBuf response = context.alloc().buffer(8);
			response.writeShort(RES_CMD_STOP_COLLECTING);
			response.writeShort(RES_STATUS_NOT_PROCESSED);
			response.writeInt(0); // no body
			final ChannelFuture writeFuture = context.writeAndFlush(response);
			writeFuture.addListener(ChannelFutureListener.CLOSE);
			buffer.release();
			return;
		}
		for (int recordCount=0; recordCount<bodyLength/24; recordCount++) {
			// read logitude
			bodyBuffer.readShort(); // discard first two bytes
			short longitudeDegreePlusMinutes = bodyBuffer.readShort();
			String logitudeDegreePlusMinutesRepresentation = Short.valueOf(longitudeDegreePlusMinutes).toString();
			double longitudeDegree = Integer.valueOf(logitudeDegreePlusMinutesRepresentation.substring(0, 2)).doubleValue();
			double longitudeMinute = Integer.valueOf(logitudeDegreePlusMinutesRepresentation.substring(2)).doubleValue();
			double longitudeMinutesAfterDot = (double)bodyBuffer.readShort();
			double longitudeMinuteValue = longitudeMinute + longitudeMinutesAfterDot/10000;
			bodyBuffer.readByte(); // discard one byte
			byte positionByte = bodyBuffer.readByte();
			String positionString;
			if ('N' == (char)positionByte) {
				positionString = "N";
			} else {
				longitudeMinuteValue *= -1;
				positionString = "S";
			}
			logger.debug("Logitude degree: " + longitudeDegree + " minute: " + Double.valueOf(longitudeMinuteValue) + " flag: " + positionString);
			double longitude = longitudeDegree + longitudeMinuteValue/60;
			// read latitude
			bodyBuffer.readShort(); // discard first two bytes
			short latitudeDegreePlusMinutes = bodyBuffer.readShort();
			String latitudeDegreePlusMinutesRepresentation = Short.valueOf(latitudeDegreePlusMinutes).toString();
			double latitudeDegree;
			double latitudeMinute;
			if (latitudeDegreePlusMinutesRepresentation.length() < 5) {
				latitudeDegree = Integer.valueOf(latitudeDegreePlusMinutesRepresentation.substring(0, 2)).doubleValue();
				latitudeMinute = Integer.valueOf(latitudeDegreePlusMinutesRepresentation.substring(2)).doubleValue();
			} else {
				latitudeDegree = Integer.valueOf(latitudeDegreePlusMinutesRepresentation.substring(0, 3)).doubleValue();
				latitudeMinute = Integer.valueOf(latitudeDegreePlusMinutesRepresentation.substring(3)).doubleValue();
			}
			double latitudeMinutesAfterDot = (double)bodyBuffer.readShort();
			double latitudeMinuteValue = latitudeMinute + latitudeMinutesAfterDot/10000;
			bodyBuffer.readByte(); // discard one byte
			positionByte = bodyBuffer.readByte();
			String latitudePosition;
			if ('E' == (char)positionByte) {
				latitudePosition = "E";
			} else {
				latitudePosition = "W";
				latitudeMinuteValue *= -1;
			}
			logger.debug("Latitude degree: " + latitudeDegree + " minute: " + Double.valueOf(latitudeMinuteValue) + " flag: " + latitudePosition);
			double latitude = latitudeDegree + latitudeMinuteValue/60;
			// read timestamp
			bodyBuffer.readByte(); // discard first byte
			String timeString = String.format("%d-%d-%d %d:%d:%d", bodyBuffer.readShort(), bodyBuffer.readByte(), bodyBuffer.readByte(), bodyBuffer.readByte(), bodyBuffer.readByte(), bodyBuffer.readByte());
			DateFormat fullFormatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
			Date date;
			try {
				date = fullFormatter.parse(timeString);
			} catch (ParseException e) {
				logger.warn("invalid date format, request ignored - " + timeString);
				continue;
			} 
			logger.info("new record received, longitude:" + longitude + " latitude:" + latitude + " date:" + timeString);
			if (null == locationDocument.getStartLocation()) {
				// update with the start location
				GeoJsonPoint startLocation = new GeoJsonPoint(latitude, longitude);
				Update update = new Update();
				update.set("startLocation", startLocation);
				mongoTemplate.updateFirst(query, update, LocationDocument.class);
			}
			LocationObject locationObject = new LocationObject();
			locationObject.setOrderId(order.getOrderId());
			locationObject.setDateOfRecord(date);
			GeoJsonPoint location = new GeoJsonPoint(latitude, longitude);
			locationObject.setLocation(location);
			mongoTemplate.save(locationObject);
		}

		// continue reporting response
		buffer.skipBytes(buffer.readableBytes());
		final ByteBuf response = context.alloc().buffer(8);
		response.writeShort(RES_CMD_KEEP_GOING);
		response.writeShort(RES_STATUS_ACCEPTED);
		response.writeInt(0);
		final ChannelFuture writeFuture = context.writeAndFlush(response);
		writeFuture.addListener(ChannelFutureListener.CLOSE);
		buffer.release();
		buffer = null;
	}
	
	protected void proceedWithTemperatureRequestBody(ChannelHandlerContext context, ByteBuf buffer, DeviceOrder order) {
		Query temperatureQuery = new Query();
		Criteria temperatureIdCriteria = Criteria.where("orderId").is(order.getOrderId());
		temperatureQuery.addCriteria(temperatureIdCriteria);
		if (!mongoTemplate.exists(temperatureQuery, TemperatureDocument.class)) {
			logger.warn("invalid package, temperature document not created, connection dropped!!!");
			buffer.skipBytes(buffer.readableBytes());
			final ByteBuf response = context.alloc().buffer(8);
			response.writeShort(RES_CMD_STOP_COLLECTING);
			response.writeShort(RES_STATUS_NOT_PROCESSED);
			response.writeInt(0);
			final ChannelFuture writeFuture = context.writeAndFlush(response);
			writeFuture.addListener(ChannelFutureListener.CLOSE);
			buffer.release();
			buffer = null;
			return;
		}
		bodyBuffer = context.alloc().buffer(28);
		bodyBuffer.writeBytes(buffer, 28);
		int temperature = bodyBuffer.readInt();
		// read logitude
		bodyBuffer.readShort(); // discard first two bytes
		short longitudeDegreePlusMinutes = bodyBuffer.readShort();
		String logitudeDegreePlusMinutesRepresentation = Short.valueOf(longitudeDegreePlusMinutes).toString();
		double longitudeDegree = Integer.valueOf(logitudeDegreePlusMinutesRepresentation.substring(0, 2)).doubleValue();
		double longitudeMinute = Integer.valueOf(logitudeDegreePlusMinutesRepresentation.substring(2)).doubleValue();
		double longitudeMinutesAfterDot = (double)bodyBuffer.readShort();
		double longitudeMinuteValue = longitudeMinute + longitudeMinutesAfterDot/10000;
		bodyBuffer.readByte(); // discard one byte
		byte positionByte = bodyBuffer.readByte();
		String positionString;
		if ('N' == (char)positionByte) {
			positionString = "N";
		} else {
			longitudeMinuteValue *= -1;
			positionString = "S";
		}
		logger.debug("Logitude degree: " + longitudeDegree + " minute: " + Double.valueOf(longitudeMinuteValue) + " flag: " + positionString);
		double longitude = longitudeDegree + longitudeMinuteValue/60;
		// read latitude
		bodyBuffer.readShort(); // discard first two bytes
		short latitudeDegreePlusMinutes = bodyBuffer.readShort();
		String latitudeDegreePlusMinutesRepresentation = Short.valueOf(latitudeDegreePlusMinutes).toString();
		double latitudeDegree;
		double latitudeMinute;
		if (latitudeDegreePlusMinutesRepresentation.length() < 5) {
			latitudeDegree = Integer.valueOf(latitudeDegreePlusMinutesRepresentation.substring(0, 2)).doubleValue();
			latitudeMinute = Integer.valueOf(latitudeDegreePlusMinutesRepresentation.substring(2)).doubleValue();
		} else {
			latitudeDegree = Integer.valueOf(latitudeDegreePlusMinutesRepresentation.substring(0, 3)).doubleValue();
			latitudeMinute = Integer.valueOf(latitudeDegreePlusMinutesRepresentation.substring(3)).doubleValue();
		}
		double latitudeMinutesAfterDot = (double)bodyBuffer.readShort();
		double latitudeMinuteValue = latitudeMinute + latitudeMinutesAfterDot/10000;
		bodyBuffer.readByte(); // discard one byte
		positionByte = bodyBuffer.readByte();
		String latitudePosition;
		if ('E' == (char)positionByte) {
			latitudePosition = "E";
		} else {
			latitudePosition = "W";
			latitudeMinuteValue *= -1;
		}
		logger.debug("Latitude degree: " + latitudeDegree + " minute: " + Double.valueOf(latitudeMinuteValue) + " flag: " + latitudePosition);
		double latitude = latitudeDegree + latitudeMinuteValue/60;
		// read timestamp
		bodyBuffer.readByte(); // discard first byte
		String timeString = String.format("%d-%d-%d %d:%d:%d", bodyBuffer.readShort(), bodyBuffer.readByte(), bodyBuffer.readByte(), bodyBuffer.readByte(), bodyBuffer.readByte(), bodyBuffer.readByte());
		DateFormat fullFormatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		Date date;
		try {
			date = fullFormatter.parse(timeString);
		} catch (ParseException e) {
			logger.warn("invalid date format, request ignored - " + timeString);
			buffer.skipBytes(buffer.readableBytes());
			final ByteBuf response = context.alloc().buffer(8);
			response.writeShort(RES_CMD_KEEP_GOING);
			response.writeShort(RES_STATUS_NOT_PROCESSED);
			response.writeInt(0);
			final ChannelFuture writeFuture = context.writeAndFlush(response);
			writeFuture.addListener(ChannelFutureListener.CLOSE);
			buffer.release();
			buffer = null;
			return;
		} 
		logger.info("recording temperature data for " + deviceSN + " temperature:" + temperature + " date:" + timeString);
		Query query = temperatureDocumentByDeviceOrder(order);
		Update update = new Update();
		TemperatureObject temperatureObject = new TemperatureObject();
		temperatureObject.setTemperature(temperature);
		temperatureObject.setLongitude(latitude);
		temperatureObject.setLatitude(longitude);
		temperatureObject.setDateOfRecord(date);
		update.addToSet("temperatureObjects", temperatureObject);
		WriteResult result = mongoTemplate.updateFirst(query, update, TemperatureDocument.class);
		if (!result.isUpdateOfExisting()) {
			logger.error("failed to record temperature for order<" + order.getOrderId() + "> deviceSN:" + deviceSN + " temperature:" + temperature + " timestamp:" + timeString + " longitude:" + longitude + " latitude:" + latitude);
		}
		
		// create alert as necessary
		if ((temperature<order.getTemperatureMin().intValue()) || (temperature>order.getTemperatureMax().intValue())) {
			// admin console
			Query realtimeQuery = new Query();
			Criteria moduleCriteria = Criteria.where("module").is(PlatformRealtimeInfo.MODULE_ALERT);
			Criteria identifierCriteria = Criteria.where("identifier").is(PlatformRealtimeInfo.IDENTIFIER_VEHICLE);
			Criteria idCriteria = Criteria.where("alertLicensePlate").is(order.getLicensePlate());
			Criteria realtimeAndCriteria = new Criteria();
			realtimeAndCriteria.andOperator(moduleCriteria, identifierCriteria, idCriteria);
			realtimeQuery.addCriteria(realtimeAndCriteria);
			Update realtimeUpdate = new Update();
			realtimeUpdate.set("module", PlatformRealtimeInfo.MODULE_ALERT);
			realtimeUpdate.set("identifier", PlatformRealtimeInfo.IDENTIFIER_VEHICLE);
			realtimeUpdate.set("alertTemperature", Integer.valueOf(temperature));
			realtimeUpdate.set("alertTemperatureMin", order.getTemperatureMin());
			realtimeUpdate.set("alertTemperatureMax", order.getTemperatureMax());
			realtimeUpdate.set("alertLicensePlate", order.getLicensePlate());
			realtimeUpdate.set("alertLongitude", Double.valueOf(latitude));
			realtimeUpdate.set("alertLatitude", Double.valueOf(longitude));
			realtimeUpdate.set("alertEvent", PlatformRealtimeInfo.ALERT_EVENT_ALARMED);
			realtimeUpdate.set("creationDate", new Date());
			mongoTemplate.upsert(realtimeQuery, realtimeUpdate, PlatformRealtimeInfo.class);
		} else {
			// try remove alarm as necessary
			Query realtimeQuery = new Query();
			Criteria moduleCriteria = Criteria.where("module").is(PlatformRealtimeInfo.MODULE_ALERT);
			Criteria identifierCriteria = Criteria.where("identifier").is(PlatformRealtimeInfo.IDENTIFIER_VEHICLE);
			Criteria idCriteria = Criteria.where("alertLicensePlate").is(order.getLicensePlate());
			Criteria eventCriteria = Criteria.where("alertEvent").is(PlatformRealtimeInfo.ALERT_EVENT_ALARMED);
			Criteria realtimeAndCriteria = new Criteria();
			realtimeAndCriteria.andOperator(moduleCriteria, identifierCriteria, idCriteria, eventCriteria);
			realtimeQuery.addCriteria(realtimeAndCriteria);
			Update realtimeUpdate = new Update();
			realtimeUpdate.set("alertEvent", PlatformRealtimeInfo.ALERT_EVENT_ALARM_REMOVED);
			realtimeUpdate.set("creationDate", new Date());
			mongoTemplate.updateFirst(realtimeQuery, realtimeUpdate, PlatformRealtimeInfo.class);
		}

		// continue reporting response
		buffer.skipBytes(buffer.readableBytes());
		final ByteBuf response = context.alloc().buffer(8);
		response.writeShort(RES_CMD_KEEP_GOING);
		if (result.isUpdateOfExisting()) {
			response.writeShort(RES_STATUS_ACCEPTED);
		} else {
			response.writeShort(RES_STATUS_NOT_PROCESSED);
		}
		response.writeInt(0);
		final ChannelFuture writeFuture = context.writeAndFlush(response);
		writeFuture.addListener(ChannelFutureListener.CLOSE);
		buffer.release();
		buffer = null;
	}
	
	protected void proceedWithHumidityRequestBody(ChannelHandlerContext context, ByteBuf buffer, DeviceOrder order) {
		Query humidityQuery = new Query();
		Criteria humidityIdCriteria = Criteria.where("orderId").is(order.getOrderId());
		humidityQuery.addCriteria(humidityIdCriteria);
		if (!mongoTemplate.exists(humidityQuery, HumidityDocument.class)) {
			logger.warn("invalid package, humidity document not created, connection dropped!!!");
			buffer.skipBytes(buffer.readableBytes());
			final ByteBuf response = context.alloc().buffer(8);
			response.writeShort(RES_CMD_STOP_COLLECTING);
			response.writeShort(RES_STATUS_NOT_PROCESSED);
			response.writeInt(0);
			final ChannelFuture writeFuture = context.writeAndFlush(response);
			writeFuture.addListener(ChannelFutureListener.CLOSE);
			buffer.release();
			buffer = null;
			return;
		}
		bodyBuffer = context.alloc().buffer(28);
		bodyBuffer.writeBytes(buffer, 28);
		int humidity = bodyBuffer.readInt();
		// read logitude
		bodyBuffer.readShort(); // discard first two bytes
		short longitudeDegreePlusMinutes = bodyBuffer.readShort();
		String logitudeDegreePlusMinutesRepresentation = Short.valueOf(longitudeDegreePlusMinutes).toString();
		double longitudeDegree = Integer.valueOf(logitudeDegreePlusMinutesRepresentation.substring(0, 2)).doubleValue();
		double longitudeMinute = Integer.valueOf(logitudeDegreePlusMinutesRepresentation.substring(2)).doubleValue();
		double longitudeMinutesAfterDot = (double)bodyBuffer.readShort();
		double longitudeMinuteValue = longitudeMinute + longitudeMinutesAfterDot/10000;
		bodyBuffer.readByte(); // discard one byte
		byte positionByte = bodyBuffer.readByte();
		String positionString;
		if ('N' == (char)positionByte) {
			positionString = "N";
		} else {
			longitudeMinuteValue *= -1;
			positionString = "S";
		}
		logger.debug("Logitude degree: " + longitudeDegree + " minute: " + Double.valueOf(longitudeMinuteValue) + " flag: " + positionString);
		double longitude = longitudeDegree + longitudeMinuteValue/60;
		// read latitude
		bodyBuffer.readShort(); // discard first two bytes
		short latitudeDegreePlusMinutes = bodyBuffer.readShort();
		String latitudeDegreePlusMinutesRepresentation = Short.valueOf(latitudeDegreePlusMinutes).toString();
		double latitudeDegree;
		double latitudeMinute;
		if (latitudeDegreePlusMinutesRepresentation.length() < 5) {
			latitudeDegree = Integer.valueOf(latitudeDegreePlusMinutesRepresentation.substring(0, 2)).doubleValue();
			latitudeMinute = Integer.valueOf(latitudeDegreePlusMinutesRepresentation.substring(2)).doubleValue();
		} else {
			latitudeDegree = Integer.valueOf(latitudeDegreePlusMinutesRepresentation.substring(0, 3)).doubleValue();
			latitudeMinute = Integer.valueOf(latitudeDegreePlusMinutesRepresentation.substring(3)).doubleValue();
		}
		double latitudeMinutesAfterDot = (double)bodyBuffer.readShort();
		double latitudeMinuteValue = latitudeMinute + latitudeMinutesAfterDot/10000;
		bodyBuffer.readByte(); // discard one byte
		positionByte = bodyBuffer.readByte();
		String latitudePosition;
		if ('E' == (char)positionByte) {
			latitudePosition = "E";
		} else {
			latitudePosition = "W";
			latitudeMinuteValue *= -1;
		}
		logger.debug("Latitude degree: " + latitudeDegree + " minute: " + Double.valueOf(latitudeMinuteValue) + " flag: " + latitudePosition);
		double latitude = latitudeDegree + latitudeMinuteValue/60;
		// read timestamp
		bodyBuffer.readByte(); // discard first byte
		String timeString = String.format("%d-%d-%d %d:%d:%d", bodyBuffer.readShort(), bodyBuffer.readByte(), bodyBuffer.readByte(), bodyBuffer.readByte(), bodyBuffer.readByte(), bodyBuffer.readByte());
		DateFormat fullFormatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
		Date date;
		try {
			date = fullFormatter.parse(timeString);
		} catch (ParseException e) {
			logger.warn("invalid date format, request ignored - " + timeString);
			buffer.skipBytes(buffer.readableBytes());
			final ByteBuf response = context.alloc().buffer(8);
			response.writeShort(RES_CMD_KEEP_GOING);
			response.writeShort(RES_STATUS_NOT_PROCESSED);
			response.writeInt(0);
			final ChannelFuture writeFuture = context.writeAndFlush(response);
			writeFuture.addListener(ChannelFutureListener.CLOSE);
			buffer.release();
			buffer = null;
			return;
		} 
		logger.info("recording humidity data for " + deviceSN + " humidity:" + humidity + " date:" + fullFormatter.format(date));
		Query query = humidityDocumentByDeviceOrder(order);
		Update update = new Update();
		HumidityObject humidityObject = new HumidityObject();
		humidityObject.setHumidity(humidity);
		humidityObject.setLongitude(latitude);
		humidityObject.setLatitude(longitude);
		humidityObject.setDateOfRecord(date);
		update.addToSet("humidityObjects", humidityObject);
		WriteResult result = mongoTemplate.updateFirst(query, update, HumidityDocument.class);
		if (!result.isUpdateOfExisting()) {
			logger.error("failed to record humidity for order<" + order.getOrderId() + "> deviceSN:" + deviceSN + " humidity:" + humidity + " timestamp:" + timeString + " longitude:" + longitude + " latitude:" + latitude);
		}

		// continue reporting response
		buffer.skipBytes(buffer.readableBytes());
		final ByteBuf response = context.alloc().buffer(8);
		response.writeShort(RES_CMD_KEEP_GOING);
		if (result.isUpdateOfExisting()) {
			response.writeShort(RES_STATUS_ACCEPTED);
		} else {
			response.writeShort(RES_STATUS_NOT_PROCESSED);
		}
		response.writeInt(0);
		final ChannelFuture writeFuture = context.writeAndFlush(response);
		writeFuture.addListener(ChannelFutureListener.CLOSE);
		buffer.release();
		buffer = null;
	}
	
	protected Query locationDocumentByDeviceOrder(DeviceOrder deviceOrder) {
		Query query = new Query();
		Criteria idCriteria = Criteria.where("orderId").is(deviceOrder.getOrderId());
		Criteria dateCriteria = Criteria.where("recordingEnd").exists(false);
		Criteria andCriteria = new Criteria();
		andCriteria.andOperator(idCriteria, dateCriteria);
		query.addCriteria(andCriteria);
		
		return query;
	}

	protected Query humidityDocumentByDeviceOrder(DeviceOrder deviceOrder) {
		Query query = new Query();
		Criteria idCriteria = Criteria.where("orderId").is(deviceOrder.getOrderId());
		Criteria dateCriteria = Criteria.where("recordingEnd").exists(false);
		Criteria andCriteria = new Criteria();
		andCriteria.andOperator(idCriteria, dateCriteria);
		query.addCriteria(andCriteria);
		
		return query;
	}
	
	protected Query temperatureDocumentByDeviceOrder(DeviceOrder deviceOrder) {
		Query query = new Query();
		Criteria idCriteria = Criteria.where("orderId").is(deviceOrder.getOrderId());
		Criteria dateCriteria = Criteria.where("recordingEnd").exists(false);
		Criteria andCriteria = new Criteria();
		andCriteria.andOperator(idCriteria, dateCriteria);
		query.addCriteria(andCriteria);
		
		return query;
	}
	
	protected boolean shouldCollectDeviceData(String deviceSN) {
		return false;
	}
	
	public ByteBuf getHeaderBuffer() {
		return headerBuffer;
	}

	public void setHeaderBuffer(ByteBuf headerBuffer) {
		this.headerBuffer = headerBuffer;
	}

	public ByteBuf getBodyBuffer() {
		return bodyBuffer;
	}

	public void setBodyBuffer(ByteBuf bodyBuffer) {
		this.bodyBuffer = bodyBuffer;
	}

	private RedisTemplate<String, Object> redisTemplate;
	
	private MongoTemplate mongoTemplate;
	
	private static final Logger logger = LoggerFactory.getLogger(CollectorInboundAdapter.class);
}
